package org.egomsl.mw.mapred;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Progressable;
import org.egomsl.mw.HadoopRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

public class ABaseOutputFormat implements OutputFormat<BytesWritable, HadoopRecord> {
    private static Logger log = LoggerFactory.getLogger(ABaseRecordWriter.class);

    public static String ABASE_TABLE_PATH_KEY = "abase.mapred.output.table.dir";

    public RecordWriter<BytesWritable, HadoopRecord>
                getRecordWriter(FileSystem ignored, JobConf job,
                                       String name, Progressable progress)
            throws IOException {
        String tablePath = job.get(ABASE_TABLE_PATH_KEY);
        if(tablePath != null) {
            Path table = new Path(tablePath);
            Path regionPath = new Path(table, name);
            return new ABaseRecordWriter(ignored, regionPath);
        }else {
            log.error("No output path specified by '"+ABASE_TABLE_PATH_KEY+"'");
        }
        return null;
    }

    public void checkOutputSpecs(FileSystem ignored, JobConf job) throws IOException {

    }

    public static void setOutputPath(Job job, String path) {
        job.getConfiguration().set(ABASE_TABLE_PATH_KEY, path);
    }

}
